package com.xf.day06
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Dataset, RelationalGroupedDataset, Row, SparkSession}
import java.util
import org.apache.spark.sql.functions._
import java.util.List
import org.apache.spark.sql.functions.{asc, col, expr}

object TestDataFrame2RDD {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("testDataFrame")
      .setMaster("local[*]")

    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    // 从 SparkSession 获取 SparkContext
    val sc = spark.sparkContext
    sc.setLogLevel("ERROR")

    val df1: DataFrame = spark
      .read
      .format("json")
      .load("D:\\workRecord\\spark_project20250924\\data\\student2.json")




//    df1.groupBy("JavaStructure")
//      .mean("oracle")
//      .show()

    // df1.show()
    // 分组聚合
    // df1.groupBy("JavaStructure")
    //   .max("dataStructure", "oracle")
    //   .show()

    // df1.groupBy("JavaStructure")
    //   .sum("oracle")
    //   .show()

    // 筛选数据
    // df1.where("name='赵敏'").show()
    // df1.where(col("name").===("赵敏")).show()

    // 筛选数据
    // df1.filter("name='赵敏'").show()
    // df1.filter( col("name").===("赵敏") ).show()

    // 筛选指定的列
    // df1.select("name", "oracle", "id").show(3, false)
    // df1.select( df1("name"), df1("id")).show(5, false)

    // df1.selectExpr("name", "name as dogname", "upper(name)", "id").show()

    // val column = df1.col("name")
    // println(column) // name

    // val column1 = df1.apply("name")
    // println(column1) // name

    // 删除其中的 id 列
    // df1.drop("id").show(10, false)

    // 获取前 3行数据
    // df1.limit(3).show()

    // 排序方式一
    // 必须导入这个隐式转换包！
    import spark.implicits._
     df1.orderBy($"id".desc, $"name".asc).show(10)

    // 排序方式二
     df1.sort( df1("name").asc ).show()

    // 按照分区排序好的 DataFrame 对象
    // df1.sortWithinPartitions("id").show()

    // 汇总 --> 先进行聚合, 后进行求和
//    df1.groupBy("JavaStructure").count().show()

    // 分组求最大值
   //  val dataset : RelationalGroupedDataset = df1.groupBy("JavaStructure")
   //  println(dataset.toString())

  }

  private def testGroupBy(): Unit = {
    val conf = new SparkConf()
      .setAppName("testDataFrame")
      .setMaster("local[*]")

    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    // 从 SparkSession 获取 SparkContext
    val sc = spark.sparkContext
    sc.setLogLevel("ERROR")

    val df1: DataFrame = spark
      .read
      .format("json")
      .load("D:\\workRecord\\spark_project20250924\\data\\student2.json")

    // 排序方式一
    // 必须导入这个隐式转换包！
    import spark.implicits._
    // 分组求最大值
    val result = df1
      .groupBy("JavaStructure")
      .agg(
        collect_list(struct("id", "name", "dataStructure", "oracle")) as "group_data"
      )
      // 将 group_data 数组展开为多行
      .withColumn("group_data", explode($"group_data"))

    result.show(false)
  }

  private def showAndFirstAndSoAnd(): Unit = {
    val conf = new SparkConf()
      .setAppName("testDataFrame")
      .setMaster("local[*]")

    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    // 从 SparkSession 获取 SparkContext
    val sc = spark.sparkContext
    sc.setLogLevel("ERROR")

    val df1: DataFrame = spark
      .read
      .format("json")
      .load("D:\\workRecord\\spark_project20250924\\data\\student1.json")

    df1.show()

    val cnt = df1.count()
    println("数据的行数为:" + cnt)

    println("返回第一条记录:" + df1.first())

    val rows = df1.head(2).toBuffer
    println("返回前2行中的数据: " + rows)

    val takeBuffer = df1.take(2).toBuffer
    println("返回take的前2行数据 " + takeBuffer)

    val rowList: util.List[Row] = df1.takeAsList(2)
    println("获取前2行中的数据 " + rowList)

    //    testSaveAsFile()
    val DataValue: Dataset[Row] = df1.distinct()
    DataValue.show(false)

    val df1Row: Dataset[Row] = df1.dropDuplicates("Java程序设计")
    df1Row.show(false)
  }

  private def testSaveAsFile(): Unit = {
    val conf = new SparkConf()
      .setAppName("testDataFrame")
      .setMaster("local[*]")

    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    val df1 = spark
      .read
      .format("json")
      .load("D:\\workRecord\\spark_project20250924\\data\\student1.json")

    // df1.printSchema()
    println("=======================================>")

    // df1.show()
    // df1.show(5) // 显示 5 条数据
    df1.show(5, true)

    val rows: Array[Row] = df1.collect()
    val buffer = rows.toBuffer
    println(buffer)

    println("============== 第2个 =========================>")
    val rows1: util.List[Row] = df1.collectAsList()
    println(rows1)

    // 删除已存在的输出目录
    val outputPath = "D:\\workRecord\\spark_project20250924\\data\\student_result"
    val hadoopConf = spark.sparkContext.hadoopConfiguration
    val fs = FileSystem.get(hadoopConf)
    val path = new Path(outputPath)

    if (fs.exists(path)) {
      fs.delete(path, true) // true 表示递归删除
      println(s"已删除已存在的目录: $outputPath")
    }

    df1.rdd.saveAsTextFile(s"file:///${outputPath}")
  }
}

/**
 * root
 * |-- Java程序设计: long (nullable = true)
 * |-- oracle: long (nullable = true)
 * |-- 姓名: string (nullable = true)
 * |-- 学号: string (nullable = true)
 * |-- 数据结构: long (nullable = true)
 *
 * =======================================>
 *
 * +------------+------+----+--------+--------+
 * |Java程序设计|oracle|姓名|    学号|数据结构|
 * +------------+------+----+--------+--------+
 * |          92|    78|张伟|20230001|      85|
 * |          88|    85|李娜|20230002|      90|
 * |          84|    80|王强|20230003|      78|
 * |          95|    72|刘洋|20230004|      93|
 * |          90|    88|陈静|20230005|      87|
 * |          82|    70|黄磊|20230006|      76|
 * |          89|    83|赵敏|20230007|      91|
 * |          94|    77|周杰|20230008|      84|
 * |          86|    90|吴芳|20230009|      88|
 * |          81|    75|徐磊|20230010|      79|
 * |          93|    86|孙丽|20230011|      92|
 * |          87|    74|胡波|20230012|      81|
 * |          91|    82|郭婷|20230013|      89|
 * |          79|    68|何勇|20230014|      77|
 * |          96|    91|高娟|20230015|      94|
 * |          85|    76|林峰|20230016|      83|
 * |          90|    80|罗鑫|20230017|      86|
 * |          83|    71|郑凯|20230018|      75|
 * |          94|    87|谢娜|20230019|      90|
 * |          88|    79|宋佳|20230020|      82|
 * +------------+------+----+--------+--------+
 *
 *
 * =======================================>
 *    保存到本地文件后，得到的结果内容为:
 *   student_result/          # 输出目录
 *   ├── .SUCCESS.crc     # 校验文件（标记作业成功完成）
 *   ├── .part-00000.crc  # 校验文件（数据文件的校验）
 *   ├── _SUCCESS         # 成功标记文件
 *   └── part-00000       # 实际数据文件
 *
 */
